package com.iotechn.iot.device.config;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.baomidou.mybatisplus.mapper.Wrapper;
import com.iotechn.iot.commons.entity.exception.ServiceException;
import com.iotechn.iot.device.Const;
import com.iotechn.iot.device.entity.DeviceDo;
import com.iotechn.iot.device.entity.DeviceParamDo;
import com.iotechn.iot.device.entity.model.DeviceParamValueModel;
import com.iotechn.iot.device.entity.model.IotExchangeModel;
import com.iotechn.iot.device.mapper.DeviceMapper;
import com.iotechn.iot.device.mapper.DeviceParamMapper;
import com.iotechn.iot.executor.api.ExecutionOpenService;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.lang.reflect.Constructor;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;

public class PushCallback implements MqttCallback {

    private static final Logger logger = LoggerFactory.getLogger(PushCallback.class);

    private static final String DEVICE_PARAM_PREFIX = "D_PARAM_";

    private static final String DEVICE_FLUSH_PARAM_PREFIX = "D_F_PARAM_";

    private DeviceMapper deviceMapper;
    private DeviceParamMapper deviceParamMapper;
    private StringRedisTemplate stringRedisTemplate;
    private MqttClient mqttClient;
    private ExecutionOpenService executionOpenService;

    public PushCallback(DeviceMapper deviceMapper, DeviceParamMapper deviceParamMapper, StringRedisTemplate stringRedisTemplate, MqttClient mqttClient, ExecutionOpenService executionOpenService) {
        this.deviceMapper = deviceMapper;
        this.deviceParamMapper = deviceParamMapper;
        this.stringRedisTemplate = stringRedisTemplate;
        this.mqttClient = mqttClient;
        this.executionOpenService = executionOpenService;
    }

    public void connectionLost(Throwable cause) {
        if (cause != null) {
            logger.warn("[客户端断开连接]", cause);
        }
        //尝试从新连接
        try {
            mqttClient.connect();
        } catch (Exception e) {
            logger.error("重新连接失败", e);
        }
    }

    public void deliveryComplete(IMqttDeliveryToken token) {

    }

    public void messageArrived(String topic, MqttMessage message) {
        String request = new String(message.getPayload());
        logger.info("[调试]: request=" + request);
        try {
            IotExchangeModel iotExchangeModel = JSONObject.parseObject(request, IotExchangeModel.class);
            if (iotExchangeModel.getType() == IotExchangeModel.TYPE_PARAM) {
                List<DeviceDo> deviceDos = deviceMapper
                        .selectList(new EntityWrapper<DeviceDo>()
                                .eq("secret_key", iotExchangeModel.getSecretKey()));
                if (!CollectionUtils.isEmpty(deviceDos)) {
                    DeviceDo deviceDo = deviceDos.get(0);
                    //回传参数包括msg 和 name
                    //1.参数名
                    String name = iotExchangeModel.getName();
                    //2.消息
                    String msg = iotExchangeModel.getMsg();

                    Wrapper<DeviceParamDo> wrapper = new EntityWrapper<DeviceParamDo>()
                            .eq("device_id", deviceDo.getId())
                            .eq("name", name);

                    List<DeviceParamDo> deviceParamDos = deviceParamMapper.selectList(wrapper);

                    if (!CollectionUtils.isEmpty(deviceParamDos) && deviceParamDos.size() == 1) {
                        DeviceParamDo paramDo = deviceParamDos.get(0);
                        //根据过期时间选择是存到数据库还是存到数据库
                        Integer expire = paramDo.getExpire();
                        if (expire <= Const.PARAM_CACHE_DB_THRESHOLD_VALUE) {
                            DeviceParamValueModel deviceParamValueModel = new DeviceParamValueModel();
                            deviceParamValueModel.setExpire(expire);
                            deviceParamValueModel.setGmtUpdate(System.currentTimeMillis());
                            deviceParamValueModel.setName(name);
                            deviceParamValueModel.setValue(msg);
                            stringRedisTemplate.opsForValue().set(DEVICE_PARAM_PREFIX + name + deviceDo.getId(), JSONObject.toJSONString(deviceParamValueModel), expire);
                            //stringRedisTemplate.opsForValue().get(DEVICE_FLUSH_PARAM_PREFIX );
                            //将缓存写入数据库，这里主要是解决写数据库频率不能高于PARAM_CACHE_DB_THRESHOLD_VALUE
                            //TODO 做定时任务持久化？？
                        } else {
                            //在数据库进行持久化
                            DeviceParamDo deviceParamDo = new DeviceParamDo();
                            deviceParamDo.setParamValue(msg);
                            deviceParamMapper.update(deviceParamDo, wrapper);
                        }

                    } else {
                        logger.info("[调试 无效参数]:" + request);
                    }
                } else {
                    logger.info("[垃圾信息到达] request=" + request + ";topic=" + topic);
                }
            } else if (iotExchangeModel.getType() == IotExchangeModel.TYPE_INVOKE) {
                String param = iotExchangeModel.getParam().trim();
                String[] raw = param.split(" ");
                String[] args = new String[raw.length - 2];
                if (raw.length > 2) {
                    for (int i = 2; i < raw.length; i++) {
                        args[i - 2] = raw[i];
                    }
                }
                try {
                    Object o = executionOpenService.invokeGroovyMethod(iotExchangeModel.getUuid(), iotExchangeModel.getMethod(), args);
                    mqttClient.publish(iotExchangeModel.getSecretKey()
                            , new MqttMessage(("${is}" + (Const.IGNORE_PARAM_LIST.contains(o.getClass()) ?
                                    o.toString() : JSONObject.toJSONString(o)) + "${ie}").getBytes("utf-8")));
                } catch (ServiceException e) {
                    mqttClient.publish(iotExchangeModel.getSecretKey()
                            , new MqttMessage(("${is}error!" + e.getMessage() + "${ie}").getBytes("utf-8")));
                }
            }

        } catch (Exception e) {
            logger.error("[消息处理] 异常 request=" + request, e);
        }

    }
}